package com.gjy.disruptor.db2;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.sql.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-24 20:52:37
 */
public class UserAttentionDisruptor {

    public static void main(String[] args) throws SQLException {
        Connection connection = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/python_db?rewriteBatchedStatements=true&cachePrepStmts=true&prepStmtCacheSize=250&prepStmtCacheSqlLimit=2048",
                "root", "Oracle@12"
        );

        int ringBufferSize = 1024 * 1024;
        Disruptor<UserAttentionEntity> disruptor = new Disruptor<>(new UserAttentionFactory(), ringBufferSize, Executors.defaultThreadFactory(),
                ProducerType.SINGLE, new YieldingWaitStrategy());
        UserAttentionConsumer consumer = new UserAttentionConsumer(connection);
        disruptor.handleEventsWith(consumer);
        disruptor.start();

        ExecutorService pool = Executors.newFixedThreadPool(3);
        pool.submit(() -> {
            try {
                PreparedStatement ps = connection.prepareStatement("select ua_id,user_id,user_attention_id from user_attention");
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    long uaId = rs.getLong("ua_id");
                    long userId = rs.getLong("user_id");
                    long userAttentionId = rs.getLong("user_attention_id");
                    UserAttentionEntity attention = new UserAttentionEntity();
                    attention.setUaId(uaId);
                    attention.setUserId(userId);
                    attention.setUserAttentionId(userAttentionId);
                    RingBuffer<UserAttentionEntity> ringBuffer = disruptor.getRingBuffer();
                    ringBuffer.publishEvent(new UserAttentionTranslator(attention));
                }
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        });
    }

}
